/*
 * Copyright 2025 Redpanda Data, Inc.
 *
 * Licensed as a Redpanda Enterprise file under the Redpanda Community
 * License (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
 */
#pragma once

#include "cloud_topics/level_one/common/object_id.h"
#include "cloud_topics/level_one/metastore/offset_interval_set.h"
#include "cloud_topics/level_one/metastore/state_update.h"
#include "model/fundamental.h"
#include "serde/envelope.h"
#include "serde/rw/enum.h"
#include "serde/rw/envelope.h"

#include <fmt/format.h>

namespace cloud_topics::l1::rpc {

enum class errc : int16_t {
    ok = 0,
    incorrect_partition,
    timed_out,
    not_leader,
    concurrent_requests,
    missing_ntp,
    out_of_range,
};

struct add_objects_reply
  : serde::
      envelope<add_objects_reply, serde::version<0>, serde::compat_version<0>> {
    auto serde_fields() { return std::tie(ec, corrected_next_offsets); }

    errc ec;

    // Corrected next offsets for subsequent add_objects_requests to try.
    // Expected to only be set on success.
    chunked_hash_map<model::topic_id_partition, kafka::offset>
      corrected_next_offsets;
};
struct add_objects_request
  : serde::envelope<
      add_objects_request,
      serde::version<0>,
      serde::compat_version<0>> {
    using resp_t = add_objects_reply;
    auto serde_fields() {
        return std::tie(metastore_partition, new_objects, new_terms);
    }

    model::partition_id metastore_partition;
    chunked_vector<new_object> new_objects;
    term_state_update_t new_terms;
};

struct replace_objects_reply
  : serde::envelope<
      replace_objects_reply,
      serde::version<0>,
      serde::compat_version<0>> {
    auto serde_fields() { return std::tie(ec); }

    errc ec;
};
struct replace_objects_request
  : serde::envelope<
      replace_objects_request,
      serde::version<0>,
      serde::compat_version<0>> {
    using resp_t = replace_objects_reply;
    auto serde_fields() {
        return std::tie(metastore_partition, new_objects, compaction_updates);
    }

    model::partition_id metastore_partition;
    chunked_vector<new_object> new_objects;
    chunked_hash_map<model::topic_id_partition, compaction_state_update>
      compaction_updates;
};

struct object_metadata
  : serde::
      envelope<object_metadata, serde::version<0>, serde::compat_version<0>> {
    auto serde_fields() {
        return std::tie(
          oid, footer_pos, object_size, first_offset, last_offset);
    }

    object_id oid;
    size_t footer_pos;
    size_t object_size;

    // The first offset (inclusive) that is within this object.
    kafka::offset first_offset;
    // The last offset (inclusive) that is within this object.
    kafka::offset last_offset;
};

struct get_first_offset_ge_reply
  : serde::envelope<
      get_first_offset_ge_reply,
      serde::version<0>,
      serde::compat_version<0>> {
    auto serde_fields() { return std::tie(ec, object); }

    errc ec;
    object_metadata object;
};
struct get_first_offset_ge_request
  : serde::envelope<
      get_first_offset_ge_request,
      serde::version<0>,
      serde::compat_version<0>> {
    using resp_t = get_first_offset_ge_reply;
    auto serde_fields() { return std::tie(tp, o); }

    model::topic_id_partition tp;
    kafka::offset o;
};

struct get_first_timestamp_ge_reply
  : serde::envelope<
      get_first_timestamp_ge_reply,
      serde::version<0>,
      serde::compat_version<0>> {
    auto serde_fields() { return std::tie(ec, object); }

    errc ec;
    object_metadata object;
};
struct get_first_timestamp_ge_request
  : serde::envelope<
      get_first_timestamp_ge_request,
      serde::version<0>,
      serde::compat_version<0>> {
    using resp_t = get_first_timestamp_ge_reply;
    auto serde_fields() { return std::tie(tp, o, ts); }

    model::topic_id_partition tp;
    kafka::offset o;
    model::timestamp ts;
};

struct get_first_offset_for_bytes_reply
  : serde::envelope<
      get_first_offset_for_bytes_reply,
      serde::version<0>,
      serde::compat_version<0>> {
    auto serde_fields() { return std::tie(offset, ec); }

    kafka::offset offset;
    errc ec{};
};

struct get_first_offset_for_bytes_request
  : serde::envelope<
      get_first_offset_for_bytes_request,
      serde::version<0>,
      serde::compat_version<0>> {
    using resp_t = get_first_offset_for_bytes_reply;
    auto serde_fields() { return std::tie(tp, size); }

    model::topic_id_partition tp;
    uint64_t size{};
};

struct get_offsets_reply
  : serde::
      envelope<get_offsets_reply, serde::version<0>, serde::compat_version<0>> {
    auto serde_fields() { return std::tie(ec, start_offset, next_offset); }

    errc ec;
    kafka::offset start_offset;
    kafka::offset next_offset;
};
struct get_offsets_request
  : serde::envelope<
      get_offsets_request,
      serde::version<0>,
      serde::compat_version<0>> {
    using resp_t = get_offsets_reply;
    auto serde_fields() { return std::tie(tp); }

    model::topic_id_partition tp;
};

struct get_compaction_info_reply
  : serde::envelope<
      get_compaction_info_reply,
      serde::version<0>,
      serde::compat_version<0>> {
    auto serde_fields() {
        return std::tie(
          ec,
          dirty_ranges,
          removable_tombstone_ranges,
          dirty_ratio,
          earliest_dirty_ts);
    }

    errc ec;
    offset_interval_set dirty_ranges;
    offset_interval_set removable_tombstone_ranges;
    double dirty_ratio;
    std::optional<model::timestamp> earliest_dirty_ts;
};
struct get_compaction_info_request
  : serde::envelope<
      get_compaction_info_request,
      serde::version<0>,
      serde::compat_version<0>> {
    using resp_t = get_compaction_info_reply;
    auto serde_fields() {
        return std::tie(tp, tombstone_removal_upper_bound_ts);
    }

    model::topic_id_partition tp;

    // Cleaned ranges with tombstones that were cleaned at or below this
    // timestamp are eligible to have tombstones entirely removed. These ranges
    // will be returned in the removable_tombstone_ranges field.
    model::timestamp tombstone_removal_upper_bound_ts;
};

struct get_term_for_offset_reply
  : serde::envelope<
      get_term_for_offset_reply,
      serde::version<0>,
      serde::compat_version<0>> {
    auto serde_fields() { return std::tie(ec, term); }

    errc ec;
    model::term_id term;
};
struct get_term_for_offset_request
  : serde::envelope<
      get_term_for_offset_request,
      serde::version<0>,
      serde::compat_version<0>> {
    using resp_t = get_term_for_offset_reply;
    auto serde_fields() { return std::tie(tp, offset); }

    model::topic_id_partition tp;
    kafka::offset offset;
};

struct get_end_offset_for_term_reply
  : serde::envelope<
      get_end_offset_for_term_reply,
      serde::version<0>,
      serde::compat_version<0>> {
    auto serde_fields() { return std::tie(ec, end_offset); }

    errc ec;
    kafka::offset end_offset;
};
struct get_end_offset_for_term_request
  : serde::envelope<
      get_end_offset_for_term_request,
      serde::version<0>,
      serde::compat_version<0>> {
    using resp_t = get_end_offset_for_term_reply;
    auto serde_fields() { return std::tie(tp, term); }

    model::topic_id_partition tp;
    model::term_id term;
};

struct set_start_offset_reply
  : serde::envelope<
      set_start_offset_reply,
      serde::version<0>,
      serde::compat_version<0>> {
    auto serde_fields() { return std::tie(ec); }

    errc ec;
};
struct set_start_offset_request
  : serde::envelope<
      set_start_offset_request,
      serde::version<0>,
      serde::compat_version<0>> {
    using resp_t = set_start_offset_reply;
    auto serde_fields() { return std::tie(tp, start_offset); }

    model::topic_id_partition tp;
    kafka::offset start_offset;
};

struct remove_topics_reply
  : serde::envelope<
      remove_topics_reply,
      serde::version<0>,
      serde::compat_version<0>> {
    auto serde_fields() { return std::tie(ec, not_removed); }

    errc ec;
    chunked_vector<model::topic_id> not_removed;
};
struct remove_topics_request
  : serde::envelope<
      remove_topics_request,
      serde::version<0>,
      serde::compat_version<0>> {
    using resp_t = remove_topics_reply;
    auto serde_fields() { return std::tie(metastore_partition, topics); }

    model::partition_id metastore_partition;
    chunked_vector<model::topic_id> topics;
};

} //  namespace cloud_topics::l1::rpc

template<>
struct fmt::formatter<cloud_topics::l1::rpc::errc> final
  : fmt::formatter<std::string_view> {
    using errc = cloud_topics::l1::rpc::errc;
    template<typename FormatContext>
    auto format(const errc& ec, FormatContext& ctx) const {
        switch (ec) {
        case errc::ok:
            return fmt::format_to(ctx.out(), "rpc::errc::ok");
        case errc::incorrect_partition:
            return fmt::format_to(ctx.out(), "rpc::errc::incorrect_partition");
        case errc::timed_out:
            return fmt::format_to(ctx.out(), "rpc::errc::timed_out");
        case errc::not_leader:
            return fmt::format_to(ctx.out(), "rpc::errc::not_leader");
        case errc::concurrent_requests:
            return fmt::format_to(ctx.out(), "rpc::errc::concurrent_requests");
        case errc::missing_ntp:
            return fmt::format_to(ctx.out(), "rpc::errc::missing_ntp");
        case errc::out_of_range:
            return fmt::format_to(ctx.out(), "rpc::errc::out_of_range");
        }
        return fmt::format_to(
          ctx.out(), "rpc::errc::unknown({})", static_cast<int>(ec));
    }
};
